package com.mjf.day1

import org.apache.flink.streaming.api.scala._

object WordCountFromBatch {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

//    env.disableOperatorChaining() // 所有的算子都是打散的，不能合并

    val stream: DataStream[String] = env.fromElements(
      "hello world",
      "hello world",
      "hello world"
    )

    val transformed: DataStream[(String, Int)] = stream
      .flatMap(_.split("\\s"))
      // 默认的共享组为default，slotSharingGroup会使后续的组都改变。不同组之间不能使用同一个slot
//      .slotSharingGroup("1")
//      .disableChaining()  // 前后算子都不能合并到同一个slot
//      .startNewChain()  // 前面算子不能合并到同一个slot
      .map((_, 1))
      .keyBy(0)
      .sum(1)

    transformed.print()

    env.execute("WordCountFromBatch")

  }
}
